/**
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements.  See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership.  The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with the License.  You may obtain
 * a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.zookeeper.server;

import com.alibaba.fastjson.JSON;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

public class NIOServerCnxnFactory extends ServerCnxnFactory implements Runnable {
    private static final Logger LOG = LoggerFactory.getLogger(NIOServerCnxnFactory.class);

    static {
        /**
         * this is to avoid the jvm bug:
         * NullPointerException in Selector.open()
         * http://bugs.sun.com/view_bug.do?bug_id=6427854
         */
        try {
            Selector.open().close();
        } catch (IOException ie) {
            LOG.error("Selector failed to open", ie);
        }
    }

    final Selector selector = Selector.open();
    /**
     * We use this buffer to do efficient socket I/O. Since there is a single sender thread per
     * NIOServerCnxn instance, we can use a member variable to only allocate it once.
     * 我们使用这个缓冲区来执行高效的socket I/O。 因为所有的发送方都是在NIOServerCnxn实例中的唯一线程，我们可以使用成员变量只分配它一次。
     * 直接内存byteBuffer
     */
    final ByteBuffer directBuffer = ByteBuffer.allocateDirect(64 * 1024);
    /**
     * 每个IP对应的所有客户端
     * ip -> Set<NIOServerCnxn>
     */
    final HashMap<InetAddress, Set<NIOServerCnxn>> ipMap = new HashMap<InetAddress, Set<NIOServerCnxn>>();
    ServerSocketChannel ss;
    int maxClientCnxns = 60;
    /**
     * 当前类实现了 Runnable。
     * thread是：创建成线程后，线程对象的引用。
     */
    Thread thread;

    /**
     * Construct a new server connection factory which will accept an unlimited number of concurrent
     * connections from each client (up to the file descriptor limits of the operating system).
     * startup(zks) must be called subsequently.
     *
     * @throws IOException
     */
    public NIOServerCnxnFactory() throws IOException {
    }

    @Override
    public void configure(InetSocketAddress addr, int maxcc) throws IOException {
        configureSaslLogin();

        // 创建一个线程：用于处理网络I/O的读写操作。
        thread = new ZooKeeperThread(this, "NIOServerCxn.Factory:" + addr);
        thread.setDaemon(true);

        // 客户端最大连接数量
        maxClientCnxns = maxcc;

        // NIO网络I/O：开启服务网络通信通道
        this.ss = ServerSocketChannel.open();
        ss.socket().setReuseAddress(true);
        LOG.info("binding to port " + addr);

        // 绑定端口:也就是在zoo.cfg中的：   clientPort=2181
        ss.socket().bind(addr);
        // 设置非阻塞
        ss.configureBlocking(false);


        // 向selector中注册 ACCEPT 事件
        ss.register(selector, SelectionKey.OP_ACCEPT);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int getMaxClientCnxnsPerHost() {
        return maxClientCnxns;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void setMaxClientCnxnsPerHost(int max) {
        maxClientCnxns = max;
    }

    @Override
    public void start() {
        // ensure thread is started once and only once
        // 应确保此线程被启动 有且只有一次。
        if (thread.getState() == Thread.State.NEW) {
            // 处理zkClient请求的线程启动。具体Runnable方法在 this#run()
            thread.start();
        }
    }

    @Override
    public void startup(ZooKeeperServer zkServer) throws IOException, InterruptedException {
        start(); // 启动线程（此线程是处理zkClient请求的线程；使用nio实现。）
        setZooKeeperServer(zkServer);
        zkServer.startdata(); // 从磁盘中加载session和data

        // 启动
        zkServer.startup();
    }

    @Override
    public InetSocketAddress getLocalAddress() {
        return (InetSocketAddress) ss.socket().getLocalSocketAddress();
    }

    @Override
    public int getLocalPort() {
        return ss.socket().getLocalPort();
    }

    private void addCnxn(NIOServerCnxn cnxn) {
        synchronized (cnxns) {
            cnxns.add(cnxn);
            synchronized (ipMap) {
                InetAddress addr = cnxn.sock.socket().getInetAddress();
                Set<NIOServerCnxn> s = ipMap.get(addr);
                if (s == null) {
                    // 一般情况下，每个host只有一个客户端连接。将initialCapacity设置为2，可以在一般情况只有一个entry的时候
                    // 减少内存使用。我们需要 设置initialCapacity为2去避免rehash when第一个entry已经被添加。
                    s = new HashSet<>(2);
                    s.add(cnxn);
                    ipMap.put(addr, s);
                } else {
                    s.add(cnxn);
                }
            }
        }
    }

    public void removeCnxn(NIOServerCnxn cnxn) {
        synchronized (cnxns) {
            // Remove the related session from the sessionMap.
            long sessionId = cnxn.getSessionId();
            if (sessionId != 0) {
                sessionMap.remove(sessionId);
            }

            // if this is not in cnxns then it's already closed
            if (!cnxns.remove(cnxn)) {
                return;
            }

            synchronized (ipMap) {
                Set<NIOServerCnxn> s = ipMap.get(cnxn.getSocketAddress());
                s.remove(cnxn);
            }

            unregisterConnection(cnxn);
        }
    }

    /**
     * 创建连接 NIOServerCnxn
     * @param clientSocket
     * @param sk
     * @return
     * @throws IOException
     */
    protected NIOServerCnxn createConnection(SocketChannel clientSocket, SelectionKey sk) throws IOException {
        return new NIOServerCnxn(zkServer, clientSocket, sk, this);
    }

    /**
     * 返回：相同IP里的客户端链接数
     *
     * @param cl
     * @return
     */
    private int getClientCnxnCount(InetAddress cl) {
        // The ipMap lock covers both the map, and its contents
        // (that is, the cnxn sets shouldn't be modified outside of
        // this lock)
        synchronized (ipMap) {
            Set<NIOServerCnxn> s = ipMap.get(cl);
            if (s == null) {
                return 0;
            }
            return s.size();
        }
    }

    /**
     * <pre>
     * 【用于处理client -> server 之间的I/O socket数据包。】
     *
     * 在 {@link NIOServerCnxnFactory#configure }方法 中使用 {@code thread = new ZooKeeperThread(this, "NIOServerCxn.Factory:" + addr); } 创建了线程。
     * thread 线程的 Runnable 逻辑为此方法。
     * </pre>
     */
    @Override
    public void run() {
        // 如果serverSocket没有被关闭，则一直执行：
        while (!ss.socket().isClosed()) {
            try {
                selector.select(1000);
                Set<SelectionKey> selected;
                synchronized (this) {
                    // 当获取事件时 把关注的keys返回
                    selected = selector.selectedKeys();
                }
                ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(selected);
                // 将List中的内容随机打乱顺序。
                Collections.shuffle(selectedList);

                for (SelectionKey selectionKey : selectedList) {
                    if (selectionKey.isAcceptable()) {
                        // server和client之间的网络socket通道
                        SocketChannel socket = ((ServerSocketChannel) selectionKey.channel()).accept();
                        // 获取到当前客户端IP
                        InetAddress inetAddress = socket.socket().getInetAddress();
                        // 当前客户端IP中的客户端链接数
                        int cnxncount = getClientCnxnCount(inetAddress);
                        if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns) {
                            LOG.warn("Too many connections from " + inetAddress + " - max is " + maxClientCnxns);
                            socket.close();// 当前客户端所在IP创建的客户端数量，超过maxClientCnxns，则关闭socket
                        } else {
                            LOG.info("Accepted socket connection from " + socket.socket().getRemoteSocketAddress());
                            socket.configureBlocking(false); // 设置非阻塞
                            SelectionKey key = socket.register(selector, SelectionKey.OP_READ); // 注册 读事件
                            // 创建与client的链接
                            NIOServerCnxn cnxn = createConnection(socket, key);
                            key.attach(cnxn); // SelectionKey的附件。注1
                            // 把cnxn缓存起来。
                            addCnxn(cnxn);
                        }

                    } else if (selectionKey.isReadable() || selectionKey.isWritable()) {
                        // 如果是‘读操作’或者‘写操作’:从附件中获取NIOServerCnxn对象，是在‘注1’中放进去的。
                        NIOServerCnxn c = (NIOServerCnxn) selectionKey.attachment();
                        c.doIO(selectionKey);

                    } else {
                        if (LOG.isDebugEnabled()) {
                            LOG.debug("Unexpected ops in select " + selectionKey.readyOps());
                        }
                    }
                }
                // channel中所有关注的事件都处理完之后清空。否则下一次循环时，会重复被执行
                selected.clear();
            } catch (RuntimeException e) {
                LOG.warn("Ignoring unexpected runtime exception", e);
            } catch (Exception e) {
                LOG.warn("Ignoring exception", e);
            }
        }

        // serverSocket被关闭：
        closeAll();
        LOG.info("NIOServerCnxn factory exited run method");
    }

    /**
     * clear all the connections in the selector
     */
    @Override
    @SuppressWarnings("unchecked")
    synchronized public void closeAll() {
        selector.wakeup();
        HashSet<NIOServerCnxn> cnxns;
        synchronized (this.cnxns) {
            cnxns = (HashSet<NIOServerCnxn>) this.cnxns.clone();
        }
        // got to clear all the connections that we have in the selector
        for (NIOServerCnxn cnxn : cnxns) {
            try {
                // don't hold this.cnxns lock as deadlock may occur
                cnxn.close();
            } catch (Exception e) {
                LOG.warn("Ignoring exception closing cnxn sessionid 0x" + Long.toHexString(cnxn.sessionId), e);
            }
        }
    }

    @Override
    public void shutdown() {
        try {
            ss.close();
            closeAll();
            thread.interrupt();
            thread.join();
            if (login != null) {
                login.shutdown();
            }
        } catch (InterruptedException e) {
            LOG.warn("Ignoring interrupted exception during shutdown", e);
        } catch (Exception e) {
            LOG.warn("Ignoring unexpected exception during shutdown", e);
        }
        try {
            selector.close();
        } catch (IOException e) {
            LOG.warn("Selector closing", e);
        }
        if (zkServer != null) {
            zkServer.shutdown();
        }
    }

    @Override
    public synchronized void closeSession(long sessionId) {
        selector.wakeup();
        closeSessionWithoutWakeup(sessionId);
    }

    private void closeSessionWithoutWakeup(long sessionId) {
        NIOServerCnxn cnxn = (NIOServerCnxn) sessionMap.remove(sessionId);
        if (cnxn != null) {
            try {
                cnxn.close();
            } catch (Exception e) {
                LOG.warn("exception during session close", e);
            }
        }
    }

    @Override
    public void join() throws InterruptedException {
        thread.join();
    }

    @Override
    public Iterable<ServerCnxn> getConnections() {
        return cnxns;
    }
}
